package com.darcytech.debezium.core.kafka.connect;

import com.darcytech.debezium.common.network.HttpClientWrap;
import com.darcytech.debezium.common.utils.ObjectMapperUtils;
import com.darcytech.debezium.core.kafka.connect.dto.ConnectorTask;
import com.darcytech.debezium.core.kafka.connect.dto.NameAndConfig;
import com.darcytech.debezium.core.kafka.connect.model.Config;
import com.darcytech.debezium.core.kafka.connect.model.Error;
import com.darcytech.debezium.core.kafka.connect.model.Task;
import com.fasterxml.jackson.core.type.TypeReference;

import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPut;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;

public final class KafkaConnectClient extends HttpClientWrap {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConnectClient.class);

    private static final long MAX_FAILURE_RETRY_TIME = 600000;

    private static final long DEFAULT_FAILURE_RETRY_TIME = 60000;

    private static final String BASE_PATH = "connectors/";

    private KafkaConnectClient(final int poolMaxTotal, final int poolMaxPerRoute, final int connectTimeout, final int connectionRequestTimeout,
                               final int socketTimeout, final List<String> addresses, final long failureRetryTime) {
        super(poolMaxTotal, poolMaxPerRoute, connectTimeout, connectionRequestTimeout, socketTimeout, addresses, failureRetryTime);
    }

    public KafkaConnectClient(final List<String> addresses, long failureRetryTime) {
        this(30, 10, 5000, 5000, 5000, addresses,
                Math.min(failureRetryTime, MAX_FAILURE_RETRY_TIME));

    }

    public KafkaConnectClient(final List<String> addresses) {
        this(addresses, DEFAULT_FAILURE_RETRY_TIME);
    }

    public void deleteConnector(String connector) throws IOException {
        execute(BASE_PATH + connector, HttpDelete.METHOD_NAME, null, null);
    }

    public List<String> getConnectors() throws IOException {
        String result = execute(BASE_PATH, HttpGet.METHOD_NAME, null, null);
        return ObjectMapperUtils.readValue(result, List.class);
    }

    public Task getTaskStatus(String connector) throws IOException {
        String result = execute(BASE_PATH + connector + "/tasks/0/status", HttpGet.METHOD_NAME, null, null);
        return ObjectMapperUtils.readValue(result, Task.class);
    }

    public ConnectorTask getConfigAndTasks(String connector) throws IOException {
        String result = execute(BASE_PATH + connector + "/tasks", HttpGet.METHOD_NAME, null, null);
        List<ConnectorTask> connectorTasks = ObjectMapperUtils.readValue(result, new TypeReference<List<ConnectorTask>>() {
        });
        if (connectorTasks == null || connectorTasks.size() == 0) {
            return null;
        }
        return connectorTasks.get(0);
    }

    public boolean putConnector(String connector, Config config) throws IOException {
        String result = execute(BASE_PATH + connector + "/config", HttpPut.METHOD_NAME, config, null);
        NameAndConfig nameAndConfig = ObjectMapperUtils.readValue(result, NameAndConfig.class);
        if (nameAndConfig.getName() == null) {
            Error error = ObjectMapperUtils.readValue(result, Error.class);
            if (error.getErrorCode() == 400) {
                LOGGER.error("config:{} is register error,errorMessage:{}", config, error.getMessage());
                return false;
            }
        }
        return true;
    }

}
